package cn.me.alphamq.broker.store;

import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.model.timewheel.TimeWheelModel;
import cn.me.alphamq.broker.core.timewheel.TimeWheelModelExecutor;
import cn.me.alphamq.broker.model.timewheel.TimeWheelSlotModel;
import cn.me.alphamq.common.util.DelayTimeUtil;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * 时间轮的管理者
 *
 * @author f
 */
public class TimeWheelModelManager {

    /**
     * 秒级的时间轮
     */
    private static final TimeWheelModel secWheel = new TimeWheelModel(TimeUnit.SECONDS);

    /**
     * 分钟级的时间轮
     */
    private static final TimeWheelModel minWheel = new TimeWheelModel(TimeUnit.MINUTES, secWheel);

    /**
     * 处理到期的延时消息的线程池，避免占用 扫描槽位的线程 的时间，固定 10 个线程
     */
    private static final ExecutorService delayMsgExecutor = Executors.newFixedThreadPool(10,
            r -> new Thread(r, "delay-msg-executor-thread-" + Thread.currentThread().getId()));

    /**
     * 秒针走过的秒数
     */
    private static final LongAdder sec = new LongAdder();

    static {
        // 开启扫描时间轮的任务
        new Thread(() -> {
            while (true) {
                try {
                    sec.increment();

                    delayMsgExecutor.submit(() -> {
                        // 处理到时间的消息
                        secWheel.exec();
                        if (sec.sum() % 60 == 0) {
                            minWheel.exec();
                        }
                    });

                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, "scan-slot-thread").start();
    }


    /**
     * 添加延迟的消息
     *
     * @param delayMessageModel 待添加的延迟消息模型
     */
    public static void add(DelayMessageModel delayMessageModel) {
        long executeTime = delayMessageModel.getExecuteTime();
        int delayTime = DelayTimeUtil.getDelayTime(executeTime);
        Object msg = delayMessageModel.getMsg();
        int msgType = delayMessageModel.getMsgType();

        // 如果延迟时间小于等于 0，则可能是因为网络延迟，直接执行即可
        if (delayTime <= 0) {
            TimeWheelModelExecutor.execute(new TimeWheelSlotModel(msg, msgType, executeTime), true);
            return;
        }

        // 计算这分钟还剩多少秒
        int remain = (int) (60 - (sec.sum() % 60));
        if (delayTime <= remain) {
            // 如果 这分钟剩余的秒数 足够 延迟的秒数，则直接将其放入秒级时间轮
            secWheel.add(delayTime, msg, msgType, executeTime);
        } else {
            // 如果 这分钟剩余的秒数 不够 延迟的秒数，则
            // 先给 延迟的秒数 减去 这分钟剩余的秒数
            delayTime -= remain;
            // 再计算 延迟的分钟数 和 剩余的秒数，不过 延迟的分钟数 需要加上 一
            int min = delayTime / 60 + 1;
            int sec = delayTime % 60;
            minWheel.add(min, msg, msgType, executeTime, sec);
        }
    }
}
